package example.reducejoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean> {

    private String fileName;

    static {
        System.out.println();
    }

    /**
     * 初始化
     */
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, TableBean>.Context context)
            throws IOException, InterruptedException {
        FileSplit fileSplit = (FileSplit) context.getInputSplit();
        fileName = fileSplit.getPath().getName();

    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, TableBean>.Context context)
            throws IOException, InterruptedException {
        String line = value.toString();
        if (fileName.contains("order")) {
            String[] split = line.split("\t");
            Text outKey = new Text();
            outKey.set(split[1]);

            TableBean bean = new TableBean();
            bean.setId(split[0]);
            bean.setPid(split[1]);
            bean.setAmount(Integer.parseInt(split[2]));
            bean.setPname("");
            bean.setFlag("order");
            context.write(outKey, bean);
        }
        else {
            String[] split = line.split("\t");
            Text outKey = new Text();
            outKey.set(split[0]);

            TableBean bean = new TableBean();
            bean.setId("");
            bean.setPid(split[0]);
            bean.setAmount(0);
            bean.setPname(split[1]);
            bean.setFlag("pd");
            context.write(outKey, bean);
        }
    }
}
